package com.dtruth.dataflow.bolt;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import com.aliasi.classify.ScoredClassification;
import com.aliasi.classify.ScoredClassifier;
import com.dtruth.dataflow.util.MyLoggerFactory;

/**
 * 频道自动划分，分类提取
 * 
 * @author 岳海亮
 * @email yhl@feheadline.com
 * @date 2014年10月27日
 */
public class AutoChannelBolt extends BaseBasicBolt {

	private static final long serialVersionUID = 1L;

	private static Logger log = MyLoggerFactory
			.getLogger(AutoChannelBolt.class);

	ScoredClassifier<CharSequence> compiledClassifier = null;
	List<String> channels = null;

	/**
	 * On create
	 */
	@SuppressWarnings("unchecked")
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		try {
			ObjectInputStream oi = new ObjectInputStream(getClass()
					.getResourceAsStream("/model/category_20150309.model"));
			compiledClassifier = (ScoredClassifier<CharSequence>) oi
					.readObject();
			oi.close();
		} catch (ClassNotFoundException | IOException e) {
			log.error(ExceptionUtils.getStackTrace(e));
		}
		channels = new ArrayList<String>() {
			private static final long serialVersionUID = 1L;
			{
				add("信托");
				add("债券");
				add("创投");
				add("地产");
				add("基金");
				add("外汇");
				add("宏观");
				add("期货");
				add("港股");
				add("理财");
				add("电商");
				add("科技");
				add("管理");
				add("美股");
				add("股票");
				add("营销");
				add("银行");
				add("黄金");
				add("保险");
			}
		};
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		Object obj = map.get("channel");
		String channel = null;
		if (obj != null) {
			channel = obj.toString().trim();
		}
		if (channel != null && !channel.equals("")
				&& channels.contains(channel)) {
			map.put("channel", channel);
			log.debug("emit:" + map.get("url"));
			collector.emit(new Values(map));
		} else {
			channel = getChannel(map);
			if (channel != null && !channel.equals("")) {
				map.put("channel", channel);
				log.debug("emit:" + map.get("url"));
				collector.emit(new Values(map));
			} else
				log.warn("autochannel is wrong:" + map.get("url"));
		}
	}

	/**
	 * 自动分类完成频道的划分
	 * 
	 * @param map
	 * @return
	 */
	private String getChannel(Map<String, Object> map) {
		String article = map.get("article") == null ? "" : map.get("article")
				.toString();
		String text = article.trim();
		if (text == null || text.length() < 5 || text.equals(""))
			return null;
		ScoredClassification channel = compiledClassifier.classify(text
				.subSequence(0, text.length()));
		String[] rs = channel.bestCategory().split("@");
		if (rs.length == 1)
			return rs[0].trim();
		if (rs.length == 2)
			return rs[0].trim() + "," + rs[1].trim();
		return null;
	}

}